package basicutil.pipeline;

import basicutil.pipeline.core.ProcessorStage;

import java.util.ArrayList;
import java.util.List;

/**
 * 使用示例
 */
public class ProcessingNodeExample {

    public static void main(String[] args) throws InterruptedException {
        // 定义数据类型
        DataType stringType = new DataType("String", String.class);
        DataType integerType = new DataType("Integer", Integer.class);
        DataType doubleType = new DataType("Double", Double.class);

        // 创建节点1：字符串处理节点
        ProcessorStage stage1 = new ProcessorStage("StringProcessor");
        stage1.addInputPort(stringType);
        stage1.addOutputPort(stringType);
        stage1.setProcessor(new StringProcessor());
        stage1.setProcessingInterval(500); // 每500ms处理一次

        // 创建节点2：数值转换节点
        ProcessorStage stage2 = new ProcessorStage("NumberConverter");
        stage2.addInputPort(stringType);
        stage2.addOutputPort(integerType);
        stage2.addOutputPort(doubleType);
        stage2.setProcessor(new NumberConverter());
        stage2.setProcessingInterval(1000); // 每1秒处理一次

        // 连接节点
        stage1.getOutputPort(0).connectTo(stage2.getInputPort(0));

        // 启动节点
        stage1.start();
        stage2.start();

        // 发送测试数据
        for (int i = 0; i < 10; i++) {
            DataMessage message = new DataMessage(stringType, "Message " + i, "TestSource");
            stage1.getInputPort(0).receiveMessage(message);
        }

        // 监控状态
        Thread monitorThread = new Thread(() -> {
            while (true) {
                System.out.printf("Node1: , processed=%d%n",
                       stage1.getProcessedCount());
                System.out.printf("Node2:  processed=%d%n",
                         stage2.getProcessedCount());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    break;
                }
            }
        });
        monitorThread.start();


        Thread.sleep(5000);

        // 停止所有节点
        stage1.stop();
        stage2.stop();
        monitorThread.interrupt();
    }

    // 示例处理逻辑：字符串处理
    static class StringProcessor implements Processor {
        @Override
        public List<DataMessage> process(List<DataMessage> inputs, ProcessorStage node) {
            StringBuilder sb = new StringBuilder();

            for (DataMessage msg : inputs) {
                sb.append(msg.getData()).append(" | ");
            }
            ArrayList<DataMessage> objects = new ArrayList<>();
            if (sb.length() > 0) {
                sb.setLength(sb.length() - 3); // 移除最后的分隔符
                objects.add(new DataMessage(DataType.STRING, "Processed: " + sb.toString(), node.getId()));

            }

            return objects;
        }
    }

    // 示例处理逻辑：数值转换
    static class NumberConverter implements Processor {
        @Override
        public List<DataMessage> process(List<DataMessage> inputs, ProcessorStage node) {
            int sum = 0;

            for (DataMessage msg : inputs) {
                String data = (String) msg.getData();
                try {
                    // 提取数字
                    String numStr = data.replaceAll("[^0-9]", "");
                    if (!numStr.isEmpty()) {
                        int num = Integer.parseInt(numStr);
                        sum += num;
                    }
                } catch (NumberFormatException e) {
                    // 忽略格式错误
                }
            }

            ArrayList<DataMessage> objects = new ArrayList<>();
            if (sum > 0) {
                objects.add(new DataMessage(DataType.INTEGER, sum, node.getId()));
                objects.add(new DataMessage(DataType.DOUBLE, sum * 1.0, node.getId()));

            }

            return objects;
        }
    }
}
